﻿// -----------------------------------------------------------------------
// <copyright file="NotificationProcessorService.cs" company="Zen Design Corp">
// TODO: Update copyright text.
// </copyright>
// -----------------------------------------------------------------------

namespace Zen.BlackBerry.Acknowledgement
{
	using System;
	using System.Collections.Generic;
	using System.Linq;
	using System.Threading;
	using System.Threading.Tasks.Dataflow;
	using Microsoft.Practices.Unity;
	using Zen.BlackBerry.Common;
	using Zen.BlackBerry.Configuration;
	using Zen.BlackBerry.Pap;
	using Zen.BlackBerry.Push.AppManagement;
	using Zen.BlackBerry.Push.Count;
	using Zen.BlackBerry.Push.Request;
	using Zen.BlackBerry.Subscription;
	using Zen.BlackBerry.Subscription.Web;

	/// <summary>
	/// Implementation of the <see cref="INotificationProcessorService"/>
	/// interface.
	/// </summary>
	public class NotificationProcessorService : INotificationProcessorService
	{
		private IPushProperties _pushProperties;
		private INotificationListenerService _listenerService;
		private IPushApplicationService _appService;
		private IPushRequestService _pushRequestService;
		private IPushRequestDetailService _pushRequestDetailService;
		private ISubscriptionService _subscriptionService;
		private IPushCountService _pushCountService;

		private BatchBlock<Notification> _batcher;
		private ActionBlock<Notification[]> _processor;
		private IDisposable _batcherToProcessorLinkage;

		/// <summary>
		/// Initializes a new instance of the <see cref="NotificationProcessorService"/> class.
		/// </summary>
		/// <param name="pushProperties">The push properties.</param>
		/// <param name="listenerService">The listener service.</param>
		/// <param name="appService">The app service.</param>
		/// <param name="pushRequestService">The push request service.</param>
		/// <param name="pushRequestDetailService">The push request detail service.</param>
		/// <param name="subscriptionService">The subscription service.</param>
		/// <param name="pushCountService">The push count service.</param>
		public NotificationProcessorService(
			[Dependency]IPushProperties pushProperties,
			[Dependency]INotificationListenerService listenerService,
			[Dependency]IPushApplicationService appService,
			[Dependency]IPushRequestService pushRequestService,
			[Dependency]IPushRequestDetailService pushRequestDetailService,
			[Dependency]ISubscriptionService subscriptionService,
			[Dependency]IPushCountService pushCountService)
		{
			_pushProperties = pushProperties;
			_listenerService = listenerService;
			_appService = appService;
			_pushRequestService = pushRequestService;
			_pushRequestDetailService = pushRequestDetailService;
			_subscriptionService = subscriptionService;
			_pushCountService = pushCountService;

			_batcher = new BatchBlock<Notification>(
				_pushProperties.AcknowledgementBatchSize);
			_processor = new ActionBlock<Notification[]>(
				(batch) => ExecuteBatch(batch),
				new ExecutionDataflowBlockOptions
				{

				});
			_batcherToProcessorLinkage = _batcher.LinkTo(
				_processor,
				new DataflowLinkOptions
				{
					Append = true,
					PropagateCompletion = true
				});
		}

		#region INotificationProcessorService Members
		/// <summary>
		/// Add a notification to the queue of notifications waiting to be
		/// processed.
		/// </summary>
		/// <param name="notification">The notification.</param>
		public void AddNotification(Notification notification)
		{
			if (!_batcher.Post(notification))
			{
				throw new PushSDKException(
					"Failed to post notification to pipeline.");
			}
		}
		#endregion

		/// <summary>
		/// Reconciles the specified notifications.
		/// </summary>
		/// <param name="notifications">The notifications.</param>
		protected virtual void Reconcile(List<Notification> notifications)
		{
			if ((notifications == null) || (!notifications.Any()))
			{
				return;
			}

			_pushRequestDetailService.UpdatePushRequestDetails(notifications);
		}

		private void ExecuteBatch(Notification[] notifications)
		{
			List<Notification> reconcileList = new List<Notification>();
			List<string> fastDeleteList = new List<string>();
			List<string> delayDeleteList = new List<string>();

			List<UnsubscribeRequest> unsubscribeList = new List<UnsubscribeRequest>();
			IDictionary<string, IList<Notification>> notificationsMap = BuildMap(notifications);

			foreach (string pushId in notificationsMap.Keys)
			{
				// Use the first notification in a block initialise state
				Notification firstNotification = notificationsMap[pushId][0];

				// First obtain the related application object
				PushApplication pushApplication = null;
				try
				{
					pushApplication = _appService.FindByAppId(firstNotification.PushApplicationId);
				}
				catch (Exception)
				{
					//log.error("Exception while looking up push application " + firstNotification.getPushApplicationId() + ": ", e);
					continue;
				}

				if (pushApplication == null)
				{
					//log.error("Push application is null for notification: " + firstNotification);
					continue;
				}

				// Determine whether this is the last notification and update
				//	the count service appropriately
				bool noAddresses = (firstNotification.Address == "*");
				bool lastNotificationReceived = false;
				if (firstNotification.AddressCount != null)
				{
					int addressCount = firstNotification.AddressCount.Value;
					if ((addressCount == 1) || (noAddresses))
					{
						lastNotificationReceived = true;
					}
					else
					{
						lastNotificationReceived = _pushCountService.Increment(
							pushId, notificationsMap[pushId].Count, addressCount);
					}

					if (lastNotificationReceived)
					{
						if (addressCount != 1)
						{
							_pushCountService.Remove(pushId);
						}

						if (pushApplication.AutoDeleteRequests)
						{
							if (addressCount == 1)
							{
								fastDeleteList.Add(pushId);
							}
							else
							{
								delayDeleteList.Add(pushId);
							}
						}
					}
				}

				for (int index = 0; index < notificationsMap[pushId].Count; ++index)
				{
					Notification resultNotification = notificationsMap[pushId][index];
					if (resultNotification.MessageState == MessageState.Pending)
					{
						continue;
					}

					// If application stores requests then we will need to
					//	reconcile later
					if (pushApplication.IsStorePushRequests)
					{
						reconcileList.Add(resultNotification);
					}

					// Make a copy of the notification to pass to listeners
					Notification notificationForListeners = CopyNotification(resultNotification);

					string address = notificationForListeners.Address;
					if ((address == null) || (address == "null"))
					{
						throw new PushSDKException("Result notification has null address");
					}

					// Validate subscription information if needed
					if ((!pushApplication.BypassSubscriptions) && (!noAddresses))
					{
						SubscriberPartial subscriber = null;
						try
						{
							if (pushApplication.IsStorePushRequests)
							{
								PushRequestDetail detail =
									_pushRequestDetailService.FindByIdAndAddress(
									resultNotification.PushId, address);

								if (detail != null)
								{
									subscriber = new SubscriberPartial
										{
											PushApplicationId = pushApplication.RimApplicationId,
											SubscriberId = detail.SubscriberId
										};
								}
								else
								{
									int pushLookupRetryDelay = _pushProperties.AcknowledgementPushLookupRetryDelay;

									//string errorMsg = string.Format(
									//    "Could not find the push request detail (pushId={0}, address={1}) in storage.",
									//    resultNotification.PushId, address);
									//log.warn(errorMsg + "  Waiting " + pushLookupRetryDelay + " milliseconds and then will try to look it up again.");

									Thread.Sleep(pushLookupRetryDelay);
									detail = _pushRequestDetailService.FindByIdAndAddress(resultNotification.PushId, address);

									if (detail != null)
									{
										subscriber = new SubscriberPartial
											{
												PushApplicationId = pushApplication.RimApplicationId,
												SubscriberId = detail.SubscriberId
											};
									}
									else
									{
										//log.error(errorMsg + "  No subscription maintenance will be performed as the subscriber " + "info could not be obtained because the push request detail could not be found.");
									}
								}
							}
							else
							{
								var subscribers = _subscriptionService
									.FindByAppIdAndAddress(
										pushApplication.RimApplicationId,
										address);

								// Use the first active candidate with a matching
								//	address
								foreach (SubscriberPartial aSubscriber in subscribers)
								{
									if (aSubscriber.Status == SubscriberStatus.Active)
									{
										subscriber = aSubscriber;
										break;
									}
								}
							}
						}
						catch (Exception e)
						{
							//log.error("Exception while looking up subscriber for application=" + pushApplication.getId() + ", address=" + address + ": ", e);
							continue;
						}

						// Setup subscriber information if we found it
						if (subscriber != null)
						{
							if (resultNotification.Code.Equals(StatusCode.AddressError))
							{
								UnsubscribeRequest unsubscribeRequest =
									new UnsubscribeRequest(
										pushApplication.RimApplicationId,
										subscriber.SubscriberId);
								unsubscribeRequest.IsSystemGenerated = true;
								unsubscribeList.Add(unsubscribeRequest);
							}

							notificationForListeners.Address = subscriber.SubscriberId;
						}
						else
						{
							notificationForListeners.Address = "unknown";
						}
					}

					// Finally determine whether this is the last notification
					notificationForListeners.IsLast =
						((lastNotificationReceived) &&
						(pushApplication.IsLastNotificationEnabled) &&
						(index == (notificationsMap[pushId].Count - 1)));

					// Pass notification to listeners via service
					_listenerService.NotifyListeners(notificationForListeners);
				}
			}

			// Perform request reconciliation
			Reconcile(reconcileList);
			reconcileList = null;

			// Unsubscribe anyone we no longer have in our subs list
			UnsubscribeSubscribers(unsubscribeList);
			unsubscribeList = null;

			// Perform fast deletes of push-requests
			if (fastDeleteList.Count > 0)
			{
				try
				{
					_pushRequestService.DeletePushRequests(fastDeleteList);
				}
				catch (PushSDKException e)
				{
					//log.error("Error auto-deleting push requests: ", e);
				}
			}
			fastDeleteList = null;

			if (delayDeleteList.Count > 0)
			{
				try
				{
					Thread.Sleep(3000);
					_pushRequestService.DeletePushRequests(delayDeleteList);
				}
				catch (PushSDKException e)
				{
					//log.error("Error auto-deleting push requests: ", e);
				}
			}
			delayDeleteList = null;
		}

		private IDictionary<string, IList<Notification>> BuildMap(Notification[] notifications)
		{
			Dictionary<string, IList<Notification>> result = new Dictionary<string, IList<Notification>>();
			foreach (Notification notification in notifications)
			{
				IList<Notification> list;
				if (!result.TryGetValue(notification.PushId, out list))
				{
					list = new List<Notification>();
					result.Add(notification.PushId, list);
				}
				list.Add(notification);
			}
			return result;
		}

		private Notification CopyNotification(Notification notification)
		{
			Notification copy = new Notification(notification, notification.PushApplicationId);
			return copy;
		}

		private void UnsubscribeSubscribers(List<UnsubscribeRequest> unsubscribeList)
		{
			foreach (UnsubscribeRequest unsubscribeRequest in unsubscribeList)
			{
				try
				{
					_subscriptionService.Unsubscribe(unsubscribeRequest);
				}
				catch (Exception e)
				{
					//log.error("CPSubscriptionFailureException caught: ", e);
				}
			}
		}
	}
}
